package pers.cz.logtail.queue.kafka;

import pers.cz.config.JefConfiguration;
import pers.cz.logtail.enums.KafkaConfigItem;
import pers.cz.logtail.queue.QueueConf;
import pers.cz.logtail.queue.QueueEnum;

/**
 * @program: JefLogTail
 * @description: kafka配置源
 * @author: Cheng Zhi
 * @create: 2023-03-15 22:34
 **/
public class KafkaSource {

    public static QueueConf getSource() {
        QueueConf queueConf = new QueueConf();
        queueConf.setAddstr(JefConfiguration.get(KafkaConfigItem.KAFKA_HOST));
        queueConf.setKeySerializer(JefConfiguration.get(KafkaConfigItem.KAFKA_KEY_SERIALIZER, "org.apache.kafka.common.serialization.StringSerializer"));
        queueConf.setValueSerializer(JefConfiguration.get(KafkaConfigItem.KAFKA_VALUE_SERIALIZER, "org.apache.kafka.common.serialization.StringSerializer"));
        queueConf.setAcks(JefConfiguration.get(KafkaConfigItem.KAFKA_ACKS, "1"));
        queueConf.setBatchSize(JefConfiguration.get(KafkaConfigItem.KAFKA_BATCH_SIZE, "16384"));
        queueConf.setBufferMemory(JefConfiguration.get(KafkaConfigItem.KAFKA_BUFFER_MEMORY, "33554432"));
        queueConf.setLingerMs(JefConfiguration.get(KafkaConfigItem.KAFKA_LINGER_MS, "10"));
        queueConf.setRetries(JefConfiguration.get(KafkaConfigItem.KAFKA_RETRIES, "2"));
        queueConf.setQueueType(QueueEnum.KAFKA);
        return queueConf;
    }
}
